package com.zs.canaldemo.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.ByteString;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.alibaba.otter.canal.protocol.Message;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Canal客户端
 *
 * @author ZhaoShuai
 * @date 2025/7/16 16:14
 */
@Component
public class CanalClient {

    @Value("${canal.server}")
    private String canalServer;

    @Value("${canal.destination}")
    private String destination;

    @PostConstruct
    public void init() {
        // 创建Canal连接器
        InetSocketAddress address = new InetSocketAddress(canalServer.split(":")[0], Integer.parseInt(canalServer.split(":")[1]));
        CanalConnector connector = CanalConnectors.newSingleConnector(
                address,
                destination,
                "",
                "");
        listenDatabaseChanges(connector);
    }

    private void listenDatabaseChanges(CanalConnector connector) {
        try {
            connector.connect();
            connector.subscribe(".*\\..*"); // 监听所有表的变更
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(100);
                long batchId = message.getId();
                if (batchId != -1) {
                    processEntries(message.getEntries());
                }
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                parseRowChange(entry.getHeader(), entry.getStoreValue());
            }
        }
    }

    private void parseRowChange(CanalEntry.Header header, ByteString data) {
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(data);
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                String eventType = rowChange.getEventType().name();
                String tableName = header.getTableName();

                // 根据事件类型处理数据
                if (eventType.equals("INSERT")) {
                    handleInsert(rowData.getAfterColumnsList(), tableName);
                } else if (eventType.equals("UPDATE")) {
                    handleUpdate(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), tableName);
                } else if (eventType.equals("DELETE")) {
                    handleDelete(rowData.getBeforeColumnsList(), tableName);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("解析Binlog数据错误", e);
        }
    }

    // 示例：处理插入事件
    private void handleInsert(List<CanalEntry.Column> columns, String tableName) {
        Map<String, String> fieldMap = columns.stream()
                .collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
        System.out.printf("INSERT -> 表: %s, 数据: %s%n", tableName, fieldMap);
        // 业务逻辑：更新缓存/发送消息等
    }

    // 示例：处理更新事件
    private void handleUpdate(List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns, String tableName) {
        Map<String, String> beforeFieldMap = beforeColumns.stream()
                .collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
        Map<String, String> afterFieldMap = afterColumns.stream()
                .collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
        System.out.printf("UPDATE -> 表: %s, 原数据: %s%n 新数据: %s%n", tableName, beforeFieldMap, afterFieldMap);
        // 业务逻辑：更新缓存/发送消息等
    }

    // 示例：处理删除事件
    private void handleDelete(List<CanalEntry.Column> columns, String tableName) {
        Map<String, String> fieldMap = columns.stream()
                .collect(Collectors.toMap(CanalEntry.Column::getName, CanalEntry.Column::getValue));
        System.out.printf("DELETE -> 表: %s, 数据: %s%n", tableName, fieldMap);
        // 业务逻辑：更新缓存/发送消息等
    }

}
